package rabbitmq

import (
	"context"
	"fmt"
	"sync"
	"time"

	"github.com/cenkalti/backoff/v4"
	amqp "github.com/rabbitmq/amqp091-go"
)

type replyToInfo struct {
	correlationID string
	replyChan     chan amqp.Delivery
	deadline      time.Time
}

var infoPool = sync.Pool{
	New: func() any {
		return &replyToInfo{
			correlationID: "",
			replyChan:     make(chan amqp.Delivery),
		}
	},
}

type TopicProducer struct {
	Name string
	// todo add confirm option with error channel
	sendExchangeOption *ExchangeOption // 关联的发送exchange信息

	replyToChan chan *replyToInfo
	closeChan   chan struct{}

	sendSession      *Session // 发送Request和Report的Session
	recvReplySession *Session // 收到Reply的Session

	recvReplyQueue amqp.Queue // 提供永久ReplyQueue供消费发送Reply消息

	recvReplyRoutingKey string

	replyConsumerName string

	errorChan chan error

	backoffOption *BackoffOption
}

type TopicConsumer struct {
	// todo add confirm option with error channel
	recvExchangeOption  *ExchangeOption // 关联的exchange信息
	replyExchangeOption *ExchangeOption // 关联的发送Reply的exchange信息

	consumerOption *ConsumerOption // 消费的配置

	recvSession      *Session // 接收消息的Session
	sendReplySession *Session // 发送Reply的Session

	errorChan chan error
}

func NewTopicProducer(name string, dialOption *DialOption, sendExchangeOption, recvExchangeOption *ExchangeOption, recvReplyQueueOption *QueueOption, recvReplyRoutingKey string, replyConsumerName string, errorChan chan error, backoffOption ...*BackoffOption) (*TopicProducer, error) {
	if len(backoffOption) == 1 && backoffOption[0].Enable {
		if backoffOption[0].MaxRetries == 0 {
			backoffOption[0].MaxRetries = 1
		}

		if backoffOption[0].Interval == 0 {
			backoffOption[0].Interval = 60 * time.Second
		}
	} else {
		backoffOption = []*BackoffOption{nil}
	}

	t := &TopicProducer{
		Name:                name,
		sendExchangeOption:  sendExchangeOption,
		recvReplyRoutingKey: recvReplyRoutingKey,
		replyConsumerName:   replyConsumerName,
		errorChan:           errorChan,
		replyToChan:         make(chan *replyToInfo, 64),
		backoffOption:       backoffOption[0],
		closeChan:           make(chan struct{}, 1),
	}

	var err error

	t.sendSession, err = topicSession(dialOption, sendExchangeOption, t.backoffOption)
	if err != nil {
		return nil, err
	}

	t.recvReplySession, err = topicSession(dialOption, recvExchangeOption, t.backoffOption)
	if err != nil {
		return nil, err
	}

	t.recvReplyQueue, err = queueDeclare(t.recvReplySession.Channel, recvReplyQueueOption)
	if err != nil {
		return nil, err
	}

	err = t.recvReplySession.Channel.QueueBind(t.recvReplyQueue.Name, recvReplyRoutingKey, recvExchangeOption.Name, recvReplyQueueOption.NoWait, nil)
	if err != nil {
		return nil, fmt.Errorf("bind receive queue error, %w", err)
	}

	err = t.recvReply()
	if err != nil {
		return nil, err
	}

	return t, nil
}

func (t *TopicProducer) Close() error {
	defer func() {
		t.closeChan <- struct{}{}
	}()

	switch {
	case t.sendSession != nil:
		err := t.sendSession.Close()
		if err != nil {
			return err
		}

		fallthrough
	case t.recvReplySession != nil:
		err := t.recvReplySession.Close()
		if err != nil {
			return err
		}
	}

	return nil
}

func (t *TopicProducer) SendReport(ctx context.Context, routingKey string, mandatory, immediate bool, msg []byte) error {
	return t.sendSession.Channel.PublishWithContext(
		ctx,
		t.sendExchangeOption.Name,
		routingKey,
		mandatory,
		immediate,
		amqp.Publishing{
			Body: msg,
		},
	)
}

func (t *TopicProducer) SendRequest(ctx context.Context, routingKey string, mandatory, immediate bool, msg []byte, timeOut time.Duration) ([]byte, error) {
	if t.backoffOption == nil {
		return t.sendRequest(ctx, routingKey, mandatory, immediate, msg, timeOut)
	}

	_backoff := newBackoff(ctx, t.backoffOption)
	var (
		re []byte
	)
	op := func() (err error) {
		re, err = t.sendRequest(ctx, routingKey, mandatory, immediate, msg, timeOut)
		return
	}

	err := backoff.Retry(op, _backoff)

	return re, err
}

func (t *TopicProducer) sendRequest(ctx context.Context, routingKey string, mandatory, immediate bool, msg []byte, timeOut time.Duration) ([]byte, error) {
	info := infoPool.Get().(*replyToInfo)
	info.correlationID = newCorrelationId()

	err := t.sendSession.Channel.PublishWithContext(
		ctx,
		t.sendExchangeOption.Name,
		routingKey,
		mandatory,
		immediate,
		amqp.Publishing{
			Body:          msg,
			ReplyTo:       t.recvReplyRoutingKey,
			CorrelationId: info.correlationID,
		},
	)
	if err != nil {
		return nil, err
	}

	info.deadline = time.Now().Add(timeOut)
	t.replyToChan <- info

	_ctx, cancel := context.WithTimeout(ctx, timeOut)
	defer cancel()

	for {
		select {
		case <-_ctx.Done():
			return nil, fmt.Errorf("get reply error, %w", _ctx.Err())
		case delivery := <-info.replyChan:
			return delivery.Body, nil
		}
	}
}

func (t *TopicProducer) recvReply() error {
	replies, err := t.recvReplySession.Channel.Consume(
		t.recvReplyQueue.Name,
		t.replyConsumerName,
		false, false, false, false, nil,
	)
	if err != nil {
		return fmt.Errorf("failed to consume reply error, %w", err)
	}

	go func() {
		var (
			replyClosed, replyToClosed bool
			idToMsg                    = make(map[string]amqp.Delivery)
			idToReply                  = make(map[string]*replyToInfo)
		)

		for {
			select {
			case <-t.closeChan:
				return
			default:
			}

			select {
			case reply, ok := <-replies:
				switch {
				case !ok || reply.CorrelationId == "":
					replyClosed = true
				case len(reply.CorrelationId) != 36:
					continue
				default:
				}

				if info, ok := idToReply[reply.CorrelationId]; ok {
					if time.Since(info.deadline) < 0 {
						info.replyChan <- reply
					}

					if err = reply.Ack(false); err != nil {
						t.errorChan <- err
					}

					infoPool.Put(info)

					delete(idToReply, reply.CorrelationId)
				} else {
					idToMsg[reply.CorrelationId] = reply
				}
			case replyTo, ok := <-t.replyToChan:
				switch {
				case !ok || replyTo == nil:
					replyClosed = true
				case len(replyTo.correlationID) != 36:
					continue
				default:

				}

				if msg, ok := idToMsg[replyTo.correlationID]; ok {
					if time.Since(replyTo.deadline) < 0 {
						replyTo.replyChan <- msg
					}

					if err = msg.Ack(false); err != nil {
						t.errorChan <- err
					}

					infoPool.Put(replyTo)

					delete(idToMsg, replyTo.correlationID)
				} else {
					idToReply[replyTo.correlationID] = replyTo
				}
			}

			if replyClosed && replyToClosed {
				return
			}
		}
	}()

	return nil
}

func NewTopicConsumer(dialOption *DialOption, recvExchangeOption, replyExchangeOption *ExchangeOption, consumerOption *ConsumerOption, errChan chan error) (*TopicConsumer, error) {
	t := &TopicConsumer{
		recvExchangeOption:  recvExchangeOption,
		replyExchangeOption: replyExchangeOption,
		consumerOption:      consumerOption,
		errorChan:           errChan,
	}

	var err error
	t.recvSession, err = topicSession(dialOption, recvExchangeOption, nil)
	if err != nil {
		return nil, err
	}

	t.sendReplySession, err = topicSession(dialOption, recvExchangeOption, nil)
	if err != nil {
		return nil, err
	}

	return t, nil
}

func (t *TopicConsumer) Close() error {
	switch {
	case t.recvSession != nil:
		err := t.recvSession.Channel.Close()
		if err != nil {
			return err
		}

		fallthrough
	case t.sendReplySession != nil:
		err := t.sendReplySession.Channel.Close()
		if err != nil {
			return err
		}
	}

	return nil
}

func (t *TopicConsumer) Recv(queueOption *QueueOption, bindingRoutingKey string) (chan *Receive, error) {
	_q, err := queueDeclare(t.recvSession.Channel, queueOption)
	if err != nil {
		return nil, fmt.Errorf("declare receive queue error, %w", err)
	}

	err = t.recvSession.Channel.QueueBind(_q.Name, bindingRoutingKey, t.recvExchangeOption.Name, false, nil)
	if err != nil {
		return nil, fmt.Errorf("bind receive queue error %w", err)
	}

	re := make(chan *Receive, 8)

	_msgs, err := consumeDeclare(t.recvSession.Channel, _q.Name, t.consumerOption)
	if err != nil {
		return nil, err
	}

	go func() {
		for i := range _msgs {
			if !t.consumerOption.AutoAck {
				if err := i.Ack(false); err != nil {
					t.errorChan <- err

					continue
				}
			}

			re <- &Receive{
				Body:          i.Body,
				CorrelationId: i.CorrelationId,
				ReplyTo:       i.ReplyTo,
			}
		}
	}()

	return re, nil
}

func (t *TopicConsumer) SendReply(ctx context.Context, received *Receive, mandatory, immediate bool, msg []byte, timeOut time.Duration) error {
	_ctx, cancel := context.WithTimeout(ctx, timeOut)
	defer cancel()

	return t.sendReplySession.Channel.PublishWithContext(
		_ctx,
		t.replyExchangeOption.Name,
		received.ReplyTo,
		mandatory,
		immediate,
		amqp.Publishing{
			CorrelationId: received.CorrelationId,
			Body:          msg,
		},
	)
}

func topicSession(dialOption *DialOption, exchangeOption *ExchangeOption, backoffOption *BackoffOption) (*Session, error) {
	if exchangeOption.Kind != TopicKind {
		return nil, fmt.Errorf("wrong exchange kind: %s, should be topic", exchangeOption.Kind)
	}

	conn, err := amqp.DialConfig(dialOption.url(), amqp.Config{
		Vhost: dialOption.VHost,
	})
	if err != nil {
		return nil, fmt.Errorf("rabbitmq/topic cannot (re)dial: %v %v: %q", err, dialOption.url(), dialOption)
	}

	if backoffOption != nil {
		_backoff := newBackoff(context.TODO(), backoffOption)

		var (
			ch *amqp.Channel
		)

		op := func() (err error) {
			ch, err = conn.Channel()

			return
		}

		err = backoff.Retry(op, _backoff)
		if err != nil {
			return nil, err
		}

		return &Session{
			Connection: conn,
			Channel:    ch,
		}, nil
	} else {
		ch, err := conn.Channel()
		if err != nil {
			return nil, fmt.Errorf("rabbitmq/topic cannot create channel: %v", err)
		}

		err = exchangeDeclare(ch, exchangeOption)
		if err != nil {
			return nil, fmt.Errorf("rabbitmq/topic cannot create channel: %v", err)
		}

		s := &Session{
			Connection: conn,
			Channel:    ch,
		}

		return s, nil
	}
}
